package cn.spark.study.core

import org.apache.spark.{SparkConf, SparkContext}

object ActionOperation {

  def main(args: Array[String]): Unit = {
    // reduce()
    // collect()
    // count()
    // take()
    // saveAsTextFile()
    countByKey()
  }

  def reduce(): Unit = {
    val conf = new SparkConf().setAppName("reduce").setMaster("local")
    val sc = new SparkContext(conf)
    val numbers = Array(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
    val count = sc.parallelize(numbers).reduce(_ + _)
    println(count)
  }

  def collect(): Unit = {
    val conf = new SparkConf().setAppName("collect").setMaster("local")
    val sc = new SparkContext(conf)
    val numbers = Array(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
    val doubleNumbers = sc.parallelize(numbers).map(num => num *2)
    val doubleNumberArray = doubleNumbers.collect();
    doubleNumberArray.foreach(num => println(num))
  }

  def count(): Unit = {
    val conf = new SparkConf().setAppName("count").setMaster("local")
    val sc = new SparkContext(conf)
    val numbers = Array(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
    val count = sc.parallelize(numbers).count()
    println(count)
  }

  def take(): Unit = {
    val conf = new SparkConf().setAppName("take").setMaster("local")
    val sc = new SparkContext(conf)
    val numbers = Array(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
    val top3Numbers = sc.parallelize(numbers).take(3)
    top3Numbers.foreach(num => println(num))
  }

  def saveAsTextFile(): Unit = {
    val conf = new SparkConf().setAppName("saveAsTextFile")
    val sc = new SparkContext(conf)
    val numbers = Array(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
    val doubleNumbers = sc.parallelize(numbers).map(num => num *2)
    doubleNumbers.saveAsTextFile("hdfs://spark1:9000/double_number_scala")
  }

  def countByKey(): Unit = {
    val conf = new SparkConf().setAppName("countByKey").setMaster("local")
    val sc = new SparkContext(conf)
    val studentArray = Array(Tuple2("class1", "leo"), Tuple2("class2", "marry"), Tuple2("class1", "jack"), Tuple2("class2", "张三"), Tuple2("class2", "李四"))
    val students = sc.parallelize(studentArray)
    val studentCounts = students.countByKey()
    println(studentCounts)
  }

}
